package monitor

import bean.BusInfo
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object RealTimeProcess {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("bus_monitor").getOrCreate()
    spark.sparkContext.setLogLevel("warn")
    import spark.implicits._
    val df: DataFrame = spark.readStream.format("kafka").options(Map(
    "kafka.bootstrap.servers" -> "linux100:9092,linux101:9092",
    "subscribe" -> "lg_bus_info"
    )).load()

    val valueDF: DataFrame = df.selectExpr("CAST(value AS STRING)")
    val ds = valueDF.as[String]
    val busInfoDs: Dataset[BusInfo] = ds.map(BusInfo(_)).filter(null != _)

    busInfoDs.writeStream.foreach(
      new MySQLWriter
    ).outputMode("append").start()

    spark.streams.awaitAnyTermination()
  }
}
